package com.imooc.bilibili.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.imooc.bilibili.constant.UserMomentsConstant;
import com.imooc.bilibili.entity.BarrageInfo;
import com.imooc.bilibili.service.BarrageService;
import com.imooc.bilibili.utils.JwtUtil;
import com.imooc.bilibili.utils.RocketMqUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@Component("webSocketService")
@ServerEndpoint("/imserver/{token}")
public class WebSocketService {

    private final Logger logger = LoggerFactory.getLogger(WebSocketService.class);

    //webSocket长连接的客户端数量
    public static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    public Session getSession() {
        return session;
    }

    public void setSession(Session session) {
        this.session = session;
    }

    //负责存储每一个来自客户端的webSocket连接
    public static final ConcurrentHashMap<String, WebSocketService> WEB_SOCKET_MAP = new ConcurrentHashMap<>();

    //服务端和客户端的会话
    public Session session;

    //以sessionId作为通信的唯一标识
    public String sessionId;

    //创建全局的ApplicationContext
    private static ApplicationContext APPLICATION_CONTEXT;

    private Long userId;

    public static void setApplicationContext(ApplicationContext applicationContext) {
        WebSocketService.APPLICATION_CONTEXT = applicationContext;
    }

    /**
     * 建立webSocket连接
     *
     * @param session
     */
    @OnOpen
    public void openConnection(Session session, @PathParam("token") String token) {
        //获取用户id
        this.userId = (Long) JwtUtil.getClaim(token).get("userId");
        //获取会话中的sessionId
        this.sessionId = session.getId();
        this.session = session;
        if (WEB_SOCKET_MAP.containsKey(sessionId)) {
            WEB_SOCKET_MAP.remove(sessionId);
            WEB_SOCKET_MAP.put(sessionId, this);
        } else {
            WEB_SOCKET_MAP.put(sessionId, this);
            ONLINE_COUNT.getAndIncrement();
        }
        logger.info("用户连接成功:" + sessionId + ",当前的在线人数为：" + ONLINE_COUNT.get());
        try {
            //后端回应前端，表示前后端连接成功
            this.sendMessage("0");
        } catch (Exception exception) {
            logger.error("连接异常");
        }
    }

    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 关闭连接---关闭或者刷新页面
     */
    @OnClose
    public void closeConeection() {
        if (WEB_SOCKET_MAP.containsKey(sessionId)) {
            WEB_SOCKET_MAP.remove(sessionId);
            ONLINE_COUNT.getAndDecrement();
        }
        logger.info("用户退出:" + sessionId + "当前在线人数:" + ONLINE_COUNT.get());
    }

    public String getSessionId() {
        return sessionId;
    }

    public void setSessionId(String sessionId) {
        this.sessionId = sessionId;
    }

    /**
     * 前后端之间消息通信
     */
    @OnMessage
    public void OnMessage(String message) {
        logger.info("用户信息:" + sessionId + ",报文:" + message);
        Timestamp timestamp = new Timestamp(System.currentTimeMillis());
        //判断message的合法性
        if (StringUtils.isNotBlank(message)) {
            try {
                //群发消息
                //遍历map数据，获取map中的value数据
                for (Map.Entry<String, WebSocketService> entry : WEB_SOCKET_MAP.entrySet()) {
                    WebSocketService webSocketService = entry.getValue();
                    //生产消息，将消息发布到MQ中
                    DefaultMQProducer producer = (DefaultMQProducer) APPLICATION_CONTEXT.getBean("barrageProducer");
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("message",message);
                    jsonObject.put("sessionId",webSocketService.getSessionId());
                    //创建消息实例
                    Message msg = new Message(UserMomentsConstant.TOPIC_BARRAGE,jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    //异步发布消息到MQ中
                    RocketMqUtil.asyncSendMsg(producer,msg);
                }
                if (null != this.userId) {
                    BarrageInfo barrageInfo = JSONObject.parseObject(message, BarrageInfo.class);
                    barrageInfo.setUserId(userId);
                    barrageInfo.setEnableState(0);
                    barrageInfo.setDeleteState(0);
                    barrageInfo.setCreateTime(timestamp);
                    barrageInfo.setUpdateTime(timestamp);
                    //多例获取弹幕业务逻辑对象
                    BarrageService barrageService = (BarrageService) APPLICATION_CONTEXT.getBean("barrageService");
                    //添加弹幕信息到数据库中
                    barrageService.asyncinsertBarrageInfoService(barrageInfo);
                    //保存弹幕信息到Redis中
                    barrageService.insertBarrageInfoServiceToRedis(barrageInfo);
                }
            } catch (Exception exception) {
                logger.error("弹幕接收异常");
                exception.printStackTrace();
            }
        }
    }

    /**
     * 发生错误
     */
    @OnError
    public void OnError(Throwable error) {

    }


    //定时任务通知前端在线观看人数
    @Scheduled(fixedRate = 5000)
    private void noticeOnlineCount() throws Exception {
        for (Map.Entry<String, WebSocketService> entry : WebSocketService.WEB_SOCKET_MAP.entrySet()) {
            WebSocketService webSocketService = entry.getValue();
            if (null != webSocketService) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("onlineCount",ONLINE_COUNT.get());
                jsonObject.put("msg","当前在线人数:"+ ONLINE_COUNT.get());
                webSocketService.sendMessage(jsonObject.toJSONString());
            }
        }
    }
}
